package org.caffinitas.ohc.gossip.webclient;

import com.google.common.base.Charsets;
import org.caffinitas.ohc.OHCache;
import org.caffinitas.ohc.linked.Util;
import org.jctools.queues.MpscLinkedQueue;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Objects;
import java.util.Set;

public class Client<K> {
    private Selector clientSelector;

    private Selector serverSelector;
    private String serverIp;
    private int serverPort;
    private SocketChannel sc;
    private boolean flag=true;
    //todo webclient 和 cache对象怎么耦合
    private OHCache cache;
    private ServerSocketChannel listenChannel;

    public Client(){};
    public Client(String IP, int port,MpscLinkedQueue<K> taskQueue,OHCache cache) throws IOException {
        this.cache=cache;
        this.createSocketChannelWittIPAddress(IP, port);
    }
    public void startServer(int port){
        intiServer(port);
        new Thread(()->{
            listen(serverSelector);
        }).start();
    }

    public  void intiServer(int port){
        try {
            serverSelector = Selector.open();
            listenChannel = ServerSocketChannel.open();
            // 绑定端口
            listenChannel.socket().bind(new InetSocketAddress(port));
            // 设置非阻塞模式
            listenChannel.configureBlocking(false);
            // 连接注册到selector
            listenChannel.register(serverSelector, SelectionKey.OP_ACCEPT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void listen(Selector selector) {
        System.out.println("服务端正在监听.........");
        try {
            while (true) {
                // 获取事件
                if (selector.select() == 0) {
                    continue;
                }
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    // 监听到不同的事件进行不同的处理
                    listenHandler(key);
                    // 删除当前key，防止重复处理
                    iterator.remove();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void listenHandler(SelectionKey key) {
        SocketChannel sc = null;
        try {
            // 连接事件处理
            if (key.isAcceptable()) {
                // 连接注册到selector
                sc = listenChannel.accept();
                sc.configureBlocking(false);
                sc.register(serverSelector, SelectionKey.OP_READ);
            }
            // 读取事件处理
            //todo 拆包处理
            if (key.isReadable()) {
                sc = (SocketChannel) key.channel();
                sc.configureBlocking(false);
                //todo 内存池
                //todo 内存协议
                ByteBuffer buffer = ByteBuffer.allocate(512);
                //todo 为什么这里要用小端接收
                buffer.order(ByteOrder.LITTLE_ENDIAN);
                System.out.println("buffer position "+buffer.position());
                //拆包---todo 异步处理--拆包完成之后数据写入ohc中需要加锁，该线程会被阻塞，应该使用异步线程；
                //todo
                // 读取消息数据--拆包
                if (sc.read(buffer) > 0) {

                    //内存拷贝
                    byte[]serizledbytes=fixedLenghFrameDecoder(buffer);
                    //反序列化
                    // 将消息转发
                    //111
                    //sendMsgToClient(msg);
                }
            }
        } catch (IOException e) {
            try {
                // 取消注册
                key.cancel();
                // 关闭通道
                sc.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }

    //拆包--固定长度
    private byte[]  fixedLenghFrameDecoder(ByteBuffer buffer){
        //按照目前缓存协议设计;从第八字节开始存储的long值代表本包的总长度;
        long len3=buffer.getInt(0);
        long len4=buffer.getInt(4);

        int size=(int)(Util.roundUpTo8(len3)+len4);
        System.out.println("数据包总长度为: "+ Util.roundUpTo8(len3)+len4);
        int length=(size) ;
        byte[] bytes=new byte[length];
        for(int i=8;i<length+8;i++){
            int index=i-8;
            bytes[index]=buffer.get(i);
        }

        String a=new String(bytes, Charsets.UTF_8);
        System.out.println(a);
        //cache.put(a);
        return bytes;
    }

    public void sendMsgToClient(String msg) throws IOException {
        msg=" from client"+msg;
        for (SelectionKey key : serverSelector.keys()) {
            Channel channel = key.channel();
            if (channel instanceof SocketChannel) {
                SocketChannel targetChannel = (SocketChannel) channel;
                // 将msg写到buffer中
                ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
                // 将buffer数据写入通道
                targetChannel.write(buffer);
            }
        }
    }


    private void createSocketChannelWittIPAddress(String IP, int port) throws IOException {
        this.serverIp = IP;
        this.serverPort = port;
        this.clientSelector =Selector.open();
        InetSocketAddress isa = new InetSocketAddress(this.serverIp, this.serverPort);
        SocketChannel sc = SocketChannel.open(isa);
        //调用open静态方法创建连接到指定主机的SocketChannel
        this.initServerChannelWithObject(sc);
    }

    private void initServerChannelWithObject(SocketChannel socketChannel) throws IOException {
        this.sc = socketChannel;
        this.sc.configureBlocking(false);
        this.sc.register(this.clientSelector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
    }



    public void start(){
        try {
            while (this.clientSelector.select() > 0) {
                //轮询所有选择器接收到的操作
                Set<SelectionKey> selectionKeys = clientSelector.selectedKeys();
                Iterator<SelectionKey> selectionKeyIte = selectionKeys.iterator();
                this.messageHandler(selectionKeyIte);
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    public void getKeySet() throws IOException {
        while (flag) {
            if(this.clientSelector.select()==0)
                continue;
            //轮询所有选择器接收到的操作
            messageHandler((clientSelector.selectedKeys().iterator()));
        }
    }

    public void messageHandler(Iterator<SelectionKey> selectionKeyIte) throws IOException {
        while (selectionKeyIte.hasNext()) {
            SelectionKey key = selectionKeyIte.next();
            //发送数据
            if (key.isWritable()) {
               //发送数据  ToDo 直接把key传过去,返回buffer
                K keyToSend=(K)this.cache.getKeyToSync();
                if(Objects.nonNull(keyToSend)){
                    sendMessage(this.cache.getTaskData(keyToSend));
                }

            }
            //接收数据
            if (key.isReadable()) {
                this.acceptMessage();
            }
            selectionKeyIte.remove();
        }
    }

    private void sendMessage(ByteBuffer databuffer) throws IOException {
        this.sc.write(databuffer);
    }

    //底层网络数据处理--预想的是这里就直接按照协议拆分成具体的entry,写入ohc中
    private void acceptMessage() throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        this.sc.read(buffer);
        buffer.flip();
        System.out.println(new String(buffer.array(),"UTF-8"));
    }
}
